package com.study.zookeeper.client.impl;

import com.alibaba.fastjson.JSONObject;
import com.study.common.base.RegisterInfo;
import com.study.common.register.IServiceRegister;
import com.study.common.utils.NetAddressUtil;
import com.study.connect.client.IConnectService;
import com.study.connect.exception.MyRpcConnectionCloseException;
import com.study.connect.exception.MyRpcTimeOutException;
import com.study.connect.message.ServiceInvokeDTO;
import com.study.zookeeper.client.IInvokeService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.ConnectException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @包名称 com.study.zookeeper.client.impl
 * @类描述 <pre></pre>
 * @作者 zhangbo
 * @创建时间 2020/5/31.
 * @版本 1.00
 * @修改记录 <pre>
 *     版本           修改人         修改日期         修改内容描述
 *     -----------------------------------------------------------
 *     1.00           zhangbo        2020/5/31.           类创建
 *     -----------------------------------------------------------
 * </pre>
 */
@Slf4j
@Component
public class InvokeServiceImpl implements IInvokeService {

    @Autowired
    private IConnectService connectService;

    @Autowired
    private IServiceRegister serviceRegister;

    @Value("${zookeeper.client.register.path:client}")
    private String clientRegisterPath;

    public static final Map<String, RegisterInfo> registerMap = new ConcurrentHashMap<>();

    public static final ReentrantLock lock = new ReentrantLock();

    @Override
    public Object invoke(ServiceInvokeDTO serviceInvokeDTO) {
        String serviceId = serviceInvokeDTO.getServiceName();
        //当服务节点宕机时，则需要重新选择负载节点
        RegisterInfo service = getService(serviceId, null);
        if (null == service) {
            throw new RuntimeException("没有服务提供者");
        }
        serviceInvokeDTO.setServiceIp(service.getServiceIp());
        serviceInvokeDTO.setServicePort(service.getServicePort());
        try {
            Object result = connectService.serviceInvoke(serviceInvokeDTO);
            return result;
        } catch (ExecutionException e) {
            log.error("远程调用异常：{}", e);
        } catch (InterruptedException e) {
            log.error("线程中断异常：{}", e);
            Thread.currentThread().interrupt();
        } catch (ConnectException | MyRpcConnectionCloseException e) {
            /*连接关闭了，本次调用不计入重试次数*/
            return reInvoke(serviceInvokeDTO, serviceInvokeDTO.getRetrys() + 1);
        } catch (MyRpcTimeOutException e) {
            /*此处需要重新路由并重试调用*/
            return reInvoke(serviceInvokeDTO, serviceInvokeDTO.getRetrys());
        } catch (Exception e) {
            log.error("RPC远程调用异常：{}", e);
        }
        return null;
    }

    /**
     * 服务重试调用
     *
     * @param serviceInvokeDTO
     * @param retries
     * @return
     */
    public Object reInvoke(ServiceInvokeDTO serviceInvokeDTO, int retries) {
        if (retries <= 0) {
            throw new RuntimeException("远程调用异常，服务连接失败");
        }
        RegisterInfo service = reSelectRegister(serviceInvokeDTO.getServiceName());
        if (null == service) {
            throw new RuntimeException("没有服务提供者");
        }
        serviceInvokeDTO.setServiceIp(service.getServiceIp());
        serviceInvokeDTO.setServicePort(service.getServicePort());
        try {
            Object result = connectService.serviceInvoke(serviceInvokeDTO);
            return result;
        } catch (ExecutionException e) {
            log.error("远程调用异常：{}", e);
        } catch (InterruptedException e) {
            log.error("线程中断异常：{}", e);
            Thread.currentThread().interrupt();
        } catch (ConnectException | MyRpcConnectionCloseException | MyRpcTimeOutException e) {
            /*此处需要重新路由并重试调用*/
            return reInvoke(serviceInvokeDTO, retries - 1);
        } catch (Exception e) {
            log.error("RPC远程调用异常：{}", e);
        }
        return null;

    }


    /**
     * 负载均衡选择
     *
     * @param serviceId
     * @return
     */
    private RegisterInfo getService(String serviceId, RegisterInfo unSelectService) {
        if (registerMap.containsKey(serviceId)) {
            if (unSelectService == null) {
                return registerMap.get(serviceId);
            } else {
                /*剔除则移除负载*/
                removeClientIpList(serviceId, unSelectService.getServiceIp(), unSelectService.getServicePort());
                registerMap.remove(serviceId);
            }
        }
        try {
            lock.lock();
            if (registerMap.containsKey(serviceId)) {
                return registerMap.get(serviceId);
            }
            List<String> registerPathInfo = serviceRegister.getRegisterPathInfo(serviceId);
            if (registerPathInfo.isEmpty()) {
                return null;
            }
            //负载选择
            if (registerPathInfo.size() > 1) {
                List<RegisterInfo> registerInfoList = new ArrayList<>(registerPathInfo.size());
                for (String serviceIp : registerPathInfo) {
                    byte[] service = serviceRegister.getRegisterInfo(serviceId + "/" + serviceIp);
                    if (null == service) {
                        continue;
                    }
                    String value = new String(service);
                    if (StringUtils.isBlank(value)) {
                        continue;
                    }
                    try {
                        RegisterInfo registerInfo = JSONObject.parseObject(value, RegisterInfo.class);
                        Set<String> clientIpList = getClientIpList(serviceId, registerInfo.getServiceIp(), registerInfo.getServicePort());
                        registerInfo.setClientIpList(clientIpList);
                        registerInfo.setClientCount(clientIpList.size());
                        if (null != unSelectService) {
                            /*排除掉需要剔除掉的链接请求*/
                            if (unSelectService.getServiceIp().equals(registerInfo.getServiceIp()) && unSelectService.getServicePort() == registerInfo.getServicePort()) {
                                continue;
                            }
                        }
                        registerInfoList.add(registerInfo);
                    } catch (Exception e) {
                        continue;
                    }
                }
                RegisterInfo minRegisterInfo = registerInfoList.get(0);
                int registerCountValue = minRegisterInfo.getClientCount() / minRegisterInfo.getWeight();
                int registerAddValue = minRegisterInfo.getClientCount() % minRegisterInfo.getWeight();
                for (int i = 1; i < registerInfoList.size(); i++) {
                    int currentCountValue = registerInfoList.get(i).getClientCount() / registerInfoList.get(i).getWeight();
                    int currentAddValue = registerInfoList.get(i).getClientCount() % registerInfoList.get(i).getWeight();
                    if (currentCountValue < registerCountValue) {
                        registerCountValue = currentCountValue;
                        registerAddValue = currentAddValue;
                        minRegisterInfo = registerInfoList.get(i);
                    } else if (currentCountValue == registerCountValue && currentAddValue < registerAddValue) {
                        registerCountValue = currentCountValue;
                        registerAddValue = currentAddValue;
                        minRegisterInfo = registerInfoList.get(i);
                    } else if (currentCountValue == registerCountValue && currentAddValue == registerAddValue && minRegisterInfo.getWeight() < registerInfoList.get(i).getWeight()) {
                        registerCountValue = currentCountValue;
                        registerAddValue = currentAddValue;
                        minRegisterInfo = registerInfoList.get(i);
                    }
                }
                /*选定，则更新负载*/
                updateClientIpList(serviceId, minRegisterInfo.getServiceIp(), minRegisterInfo.getServicePort());
                registerMap.put(serviceId, minRegisterInfo);
                return minRegisterInfo;
            } else {
                /*单节点，不需要做负载*/
                byte[] service = serviceRegister.getRegisterInfo(serviceId + "/" + registerPathInfo.get(0));
                if (null == service) {
                    return null;
                }
                String value = new String(service);
                if (StringUtils.isBlank(value)) {
                    return null;
                }
                try {
                    RegisterInfo registerInfo = JSONObject.parseObject(value, RegisterInfo.class);
                    registerMap.put(serviceId, registerInfo);
                    updateClientIpList(serviceId, registerInfo.getServiceIp(), registerInfo.getServicePort());
                    return registerInfo;
                } catch (Exception e) {
                    return null;
                }
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 重新负载分配，排除当前连接
     *
     * @param serviceId
     */
    private RegisterInfo reSelectRegister(String serviceId) {
        RegisterInfo service = getService(serviceId, null);
        return getService(serviceId, service);
    }

    /**
     * 获取客户端列表
     *
     * @param serviceId
     * @param serviceIp
     * @return
     */
    private Set<String> getClientIpList(String serviceId, String serviceIp, int port) {
        try {
            List<String> registerPathInfo = serviceRegister.getRegisterPathInfo(clientRegisterPath + "/" + serviceId + "/" + serviceIp + ":" + port);
            return new HashSet<>(registerPathInfo);
        } catch (Exception e) {
            log.error("获取客户端注册列表错误");
        }
        return new HashSet<>();
    }

    /**
     * 更新注册信息
     *
     * @param serviceId
     * @param serviceIp
     */
    private void updateClientIpList(String serviceId, String serviceIp, int port) {
        try {
            String path = clientRegisterPath + "/" + serviceId + "/" + serviceIp + ":" + port + "/" + NetAddressUtil.getServerIp();
            serviceRegister.register(path, "".getBytes());
        } catch (Exception e) {
            log.error("负载注册异常：{}", e);
        }
    }


    /**
     * 剔除注册信息
     *
     * @param serviceId
     * @param serviceIp
     */
    private void removeClientIpList(String serviceId, String serviceIp, int port) {
        try {
            String path = clientRegisterPath + "/" + serviceId + "/" + serviceIp + ":" + port;
            List<String> registerPathInfo = serviceRegister.getRegisterPathInfo(path);
            if (null == registerPathInfo || registerPathInfo.isEmpty()) {
                /*将当前地址注册至目录中*/
                return;
            }
            serviceRegister.deleteRegisterPathInfo(path + "/" + NetAddressUtil.getServerIp());
        } catch (Exception e) {
            log.error("负载移除异常：{}", e);
        }
    }
}
